package com.flink.hbase.statemanagement.managed;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * Rich Function状态示例
 * 
 * 展示如何通过RuntimeContext访问状态：
 * 1. RichMapFunction中的状态使用
 * 2. RichFlatMapFunction中的状态管理
 * 3. RichFilterFunction中的状态过滤
 * 4. RichReduceFunction中的状态聚合
 * 5. 状态TTL配置
 * 
 * @author Flink Team
 */
public class RichFunctionStateExample {
    
    /**
     * 订单事件
     */
    public static class OrderEvent {
        public String orderId;
        public String userId;
        public String productId;
        public double amount;
        public String status;
        public long timestamp;
        
        public OrderEvent() {}
        
        public OrderEvent(String orderId, String userId, String productId, 
                         double amount, String status, long timestamp) {
            this.orderId = orderId;
            this.userId = userId;
            this.productId = productId;
            this.amount = amount;
            this.status = status;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("OrderEvent{orderId='%s', userId='%s', productId='%s', amount=%.2f, status='%s', timestamp=%d}", 
                orderId, userId, productId, amount, status, timestamp);
        }
    }
    
    /**
     * 用户统计信息
     */
    public static class UserStats {
        public String userId;
        public int orderCount;
        public double totalAmount;
        public double averageAmount;
        public long lastOrderTime;
        
        public UserStats() {}
        
        public UserStats(String userId, int orderCount, double totalAmount, 
                        double averageAmount, long lastOrderTime) {
            this.userId = userId;
            this.orderCount = orderCount;
            this.totalAmount = totalAmount;
            this.averageAmount = averageAmount;
            this.lastOrderTime = lastOrderTime;
        }
        
        @Override
        public String toString() {
            return String.format("UserStats{userId='%s', orderCount=%d, totalAmount=%.2f, averageAmount=%.2f, lastOrderTime=%d}", 
                userId, orderCount, totalAmount, averageAmount, lastOrderTime);
        }
    }
    
    /**
     * 使用RichMapFunction进行状态转换
     * 计算用户订单统计信息
     */
    public static class OrderStatsMapper extends RichMapFunction<OrderEvent, UserStats> {
        
        // 订单计数状态
        private ValueState<Integer> orderCountState;
        // 总金额状态
        private ValueState<Double> totalAmountState;
        // 最后订单时间状态
        private ValueState<Long> lastOrderTimeState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            // 创建状态描述符
            ValueStateDescriptor<Integer> orderCountDescriptor = new ValueStateDescriptor<>(
                "order-count", Integer.class, 0);
            
            ValueStateDescriptor<Double> totalAmountDescriptor = new ValueStateDescriptor<>(
                "total-amount", Double.class, 0.0);
            
            ValueStateDescriptor<Long> lastOrderTimeDescriptor = new ValueStateDescriptor<>(
                "last-order-time", Long.class, 0L);
            
            // 配置状态TTL（24小时）
            StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.hours(24))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
            
            orderCountDescriptor.enableTimeToLive(ttlConfig);
            totalAmountDescriptor.enableTimeToLive(ttlConfig);
            lastOrderTimeDescriptor.enableTimeToLive(ttlConfig);
            
            // 通过RuntimeContext获取状态
            orderCountState = getRuntimeContext().getState(orderCountDescriptor);
            totalAmountState = getRuntimeContext().getState(totalAmountDescriptor);
            lastOrderTimeState = getRuntimeContext().getState(lastOrderTimeDescriptor);
        }
        
        @Override
        public UserStats map(OrderEvent order) throws Exception {
            // 获取当前状态值
            Integer currentOrderCount = orderCountState.value();
            Double currentTotalAmount = totalAmountState.value();
            Long currentLastOrderTime = lastOrderTimeState.value();
            
            // 更新状态
            int newOrderCount = currentOrderCount + 1;
            double newTotalAmount = currentTotalAmount + order.amount;
            long newLastOrderTime = Math.max(currentLastOrderTime, order.timestamp);
            
            // 保存更新后的状态
            orderCountState.update(newOrderCount);
            totalAmountState.update(newTotalAmount);
            lastOrderTimeState.update(newLastOrderTime);
            
            // 计算平均金额
            double averageAmount = newTotalAmount / newOrderCount;
            
            System.out.println(String.format("📊 用户 %s 订单统计更新: 订单数=%d, 总金额=%.2f, 平均金额=%.2f", 
                order.userId, newOrderCount, newTotalAmount, averageAmount));
            
            return new UserStats(order.userId, newOrderCount, newTotalAmount, averageAmount, newLastOrderTime);
        }
    }
    
    /**
     * 使用RichFlatMapFunction进行状态扁平化处理
     * 检测用户异常订单行为
     */
    public static class AnomalyDetector extends RichFlatMapFunction<OrderEvent, Tuple3<String, String, String>> {
        
        // 用户最近订单历史状态
        private ListState<OrderEvent> recentOrdersState;
        // 用户平均订单金额状态
        private ValueState<Double> averageAmountState;
        // 异常计数状态
        private ValueState<Integer> anomalyCountState;
        
        private static final int RECENT_ORDERS_SIZE = 10;
        private static final double ANOMALY_THRESHOLD = 3.0; // 3倍标准差
        
        @Override
        public void open(Configuration parameters) throws Exception {
            // 创建状态描述符
            ListStateDescriptor<OrderEvent> recentOrdersDescriptor = new ListStateDescriptor<>(
                "recent-orders", TypeInformation.of(OrderEvent.class));
            
            ValueStateDescriptor<Double> averageAmountDescriptor = new ValueStateDescriptor<>(
                "average-amount", Double.class, 0.0);
            
            ValueStateDescriptor<Integer> anomalyCountDescriptor = new ValueStateDescriptor<>(
                "anomaly-count", Integer.class, 0);
            
            // 配置状态TTL
            StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.hours(12))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
            
            recentOrdersDescriptor.enableTimeToLive(ttlConfig);
            averageAmountDescriptor.enableTimeToLive(ttlConfig);
            anomalyCountDescriptor.enableTimeToLive(ttlConfig);
            
            // 通过RuntimeContext获取状态
            recentOrdersState = getRuntimeContext().getListState(recentOrdersDescriptor);
            averageAmountState = getRuntimeContext().getState(averageAmountDescriptor);
            anomalyCountState = getRuntimeContext().getState(anomalyCountDescriptor);
        }
        
        @Override
        public void flatMap(OrderEvent order, Collector<Tuple3<String, String, String>> collector) throws Exception {
            // 获取最近订单历史
            List<OrderEvent> recentOrders = new ArrayList<>();
            for (OrderEvent recentOrder : recentOrdersState.get()) {
                recentOrders.add(recentOrder);
            }
            
            // 计算当前平均金额
            double currentAverage = averageAmountState.value();
            
            // 检测异常
            boolean isAnomaly = false;
            String anomalyType = "";
            
            if (recentOrders.size() >= 5) { // 需要至少5个历史订单
                // 计算标准差
                double sum = recentOrders.stream().mapToDouble(o -> o.amount).sum();
                double mean = sum / recentOrders.size();
                double variance = recentOrders.stream()
                    .mapToDouble(o -> Math.pow(o.amount - mean, 2))
                    .sum() / recentOrders.size();
                double stdDev = Math.sqrt(variance);
                
                // 检测金额异常
                if (Math.abs(order.amount - mean) > ANOMALY_THRESHOLD * stdDev) {
                    isAnomaly = true;
                    anomalyType = order.amount > mean ? "HIGH_AMOUNT" : "LOW_AMOUNT";
                }
                
                // 检测频率异常
                if (recentOrders.size() >= 3) {
                    long lastOrderTime = recentOrders.get(recentOrders.size() - 1).timestamp;
                    long timeDiff = order.timestamp - lastOrderTime;
                    if (timeDiff < 60000) { // 1分钟内连续订单
                        isAnomaly = true;
                        anomalyType = "HIGH_FREQUENCY";
                    }
                }
            }
            
            // 如果检测到异常，输出告警
            if (isAnomaly) {
                int currentAnomalyCount = anomalyCountState.value();
                anomalyCountState.update(currentAnomalyCount + 1);
                
                collector.collect(new Tuple3<>(order.userId, order.orderId, anomalyType));
                
                System.out.println(String.format("🚨 异常检测: 用户 %s 订单 %s 异常类型: %s", 
                    order.userId, order.orderId, anomalyType));
            }
            
            // 更新最近订单历史
            recentOrders.add(order);
            if (recentOrders.size() > RECENT_ORDERS_SIZE) {
                recentOrders.remove(0);
            }
            
            // 更新状态
            recentOrdersState.clear();
            recentOrdersState.addAll(recentOrders);
            
            // 更新平均金额
            double newAverage = recentOrders.stream().mapToDouble(o -> o.amount).average().orElse(0.0);
            averageAmountState.update(newAverage);
        }
    }
    
    /**
     * 使用RichFilterFunction进行状态过滤
     * 过滤重复订单
     */
    public static class DuplicateOrderFilter extends RichFilterFunction<OrderEvent> {
        
        // 已处理订单ID状态
        private ValueState<Boolean> processedOrderState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            // 创建状态描述符
            ValueStateDescriptor<Boolean> processedOrderDescriptor = new ValueStateDescriptor<>(
                "processed-order", Boolean.class, false);
            
            // 配置状态TTL（1小时）
            StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.hours(1))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
            
            processedOrderDescriptor.enableTimeToLive(ttlConfig);
            
            // 通过RuntimeContext获取状态
            processedOrderState = getRuntimeContext().getState(processedOrderDescriptor);
        }
        
        @Override
        public boolean filter(OrderEvent order) throws Exception {
            // 检查订单是否已处理
            Boolean isProcessed = processedOrderState.value();
            
            if (isProcessed) {
                System.out.println("🚫 重复订单过滤: " + order.orderId);
                return false; // 过滤掉重复订单
            } else {
                // 标记订单为已处理
                processedOrderState.update(true);
                System.out.println("✅ 新订单通过: " + order.orderId);
                return true; // 允许新订单通过
            }
        }
    }
    
    /**
     * 使用RichReduceFunction进行状态聚合
     * 聚合用户订单金额
     */
    public static class OrderAmountAggregator extends RichReduceFunction<OrderEvent> {
        
        // 聚合计数状态
        private ValueState<Integer> aggregationCountState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            // 创建状态描述符
            ValueStateDescriptor<Integer> aggregationCountDescriptor = new ValueStateDescriptor<>(
                "aggregation-count", Integer.class, 0);
            
            // 通过RuntimeContext获取状态
            aggregationCountState = getRuntimeContext().getState(aggregationCountDescriptor);
        }
        
        @Override
        public OrderEvent reduce(OrderEvent order1, OrderEvent order2) throws Exception {
            // 更新聚合计数
            Integer currentCount = aggregationCountState.value();
            aggregationCountState.update(currentCount + 1);
            
            // 创建聚合后的订单
            OrderEvent aggregatedOrder = new OrderEvent(
                order1.orderId, // 保留第一个订单ID
                order1.userId,
                order1.productId,
                order1.amount + order2.amount, // 聚合金额
                "AGGREGATED",
                Math.max(order1.timestamp, order2.timestamp) // 保留最新时间戳
            );
            
            System.out.println(String.format("🔄 订单聚合: %s + %s = %.2f (聚合次数: %d)", 
                order1.orderId, order2.orderId, aggregatedOrder.amount, currentCount + 1));
            
            return aggregatedOrder;
        }
    }
    
    /**
     * 状态清理函数
     * 演示如何清理状态
     */
    public static class StateCleaner extends RichMapFunction<OrderEvent, OrderEvent> {
        
        private ValueState<Long> lastCleanupTimeState;
        private ValueState<Integer> recordCountState;
        
        private static final long CLEANUP_INTERVAL = 60000; // 1分钟清理一次
        
        @Override
        public void open(Configuration parameters) throws Exception {
            ValueStateDescriptor<Long> lastCleanupTimeDescriptor = new ValueStateDescriptor<>(
                "last-cleanup-time", Long.class, 0L);
            
            ValueStateDescriptor<Integer> recordCountDescriptor = new ValueStateDescriptor<>(
                "record-count", Integer.class, 0);
            
            lastCleanupTimeState = getRuntimeContext().getState(lastCleanupTimeDescriptor);
            recordCountState = getRuntimeContext().getState(recordCountDescriptor);
        }
        
        @Override
        public OrderEvent map(OrderEvent order) throws Exception {
            // 更新记录计数
            Integer currentCount = recordCountState.value();
            recordCountState.update(currentCount + 1);
            
            // 检查是否需要清理状态
            Long lastCleanupTime = lastCleanupTimeState.value();
            long currentTime = System.currentTimeMillis();
            
            if (currentTime - lastCleanupTime > CLEANUP_INTERVAL) {
                // 执行状态清理
                performStateCleanup();
                lastCleanupTimeState.update(currentTime);
                
                System.out.println("🧹 状态清理完成: " + order.userId + " (处理记录数: " + recordCountState.value() + ")");
            }
            
            return order;
        }
        
        /**
         * 执行状态清理
         */
        private void performStateCleanup() {
            // 这里可以实现具体的状态清理逻辑
            // 例如：清理过期的状态、压缩状态数据等
            System.out.println("🔧 执行状态清理逻辑...");
        }
    }
    
    /**
     * 模拟订单事件源
     */
    public static class MockOrderEventSource implements SourceFunction<OrderEvent> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] userIds = {"user1", "user2", "user3", "user4", "user5"};
        private final String[] productIds = {"product1", "product2", "product3", "product4", "product5"};
        private final String[] statuses = {"CREATED", "PAID", "SHIPPED", "DELIVERED", "CANCELLED"};
        private int orderIdCounter = 1;
        
        @Override
        public void run(SourceContext<OrderEvent> ctx) throws Exception {
            while (running) {
                // 偶尔生成重复订单ID用于测试去重
                String orderId = random.nextInt(100) < 5 ? 
                    "order_" + Math.max(1, orderIdCounter - 5) : 
                    "order_" + (orderIdCounter++);
                
                OrderEvent order = new OrderEvent(
                    orderId,
                    userIds[random.nextInt(userIds.length)],
                    productIds[random.nextInt(productIds.length)],
                    50.0 + random.nextDouble() * 450.0, // 50-500随机金额
                    statuses[random.nextInt(statuses.length)],
                    System.currentTimeMillis()
                );
                
                ctx.collect(order);
                Thread.sleep(1000 + random.nextInt(2000)); // 1-3秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 运行Rich Function状态示例
     */
    public static void runRichFunctionStateExample(StreamExecutionEnvironment env) throws Exception {
        System.out.println("🚀 开始运行Rich Function状态示例...");
        
        // 创建订单事件流
        DataStream<OrderEvent> orderStream = env.addSource(new MockOrderEventSource())
            .name("订单事件源");
        
        // 1. 使用RichMapFunction计算用户统计
        DataStream<UserStats> userStatsStream = orderStream
            .keyBy(order -> order.userId)
            .map(new OrderStatsMapper())
            .name("订单统计映射");
        
        userStatsStream.print("用户统计");
        
        // 2. 使用RichFilterFunction过滤重复订单
        DataStream<OrderEvent> filteredOrderStream = orderStream
            .keyBy(order -> order.orderId)
            .filter(new DuplicateOrderFilter())
            .name("重复订单过滤");
        
        // 3. 使用RichFlatMapFunction检测异常
        DataStream<Tuple3<String, String, String>> anomalyStream = filteredOrderStream
            .keyBy(order -> order.userId)
            .flatMap(new AnomalyDetector())
            .name("异常检测");
        
        anomalyStream.print("异常告警");
        
        // 4. 使用RichReduceFunction聚合订单
        DataStream<OrderEvent> aggregatedOrderStream = filteredOrderStream
            .keyBy(order -> order.userId)
            .reduce(new OrderAmountAggregator())
            .name("订单聚合");
        
        aggregatedOrderStream.print("聚合订单");
        
        // 5. 状态清理示例
        DataStream<OrderEvent> cleanedOrderStream = filteredOrderStream
            .keyBy(order -> order.userId)
            .map(new StateCleaner())
            .name("状态清理");
        
        System.out.println("✅ Rich Function状态示例配置完成");
        System.out.println("💡 提示: 所有状态都通过RuntimeContext获取");
        System.out.println("⏰ 提示: 状态配置了TTL，会自动过期清理");
    }
    
    /**
     * 单独运行Rich Function状态示例的主方法
     */
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        
        // 配置状态后端
        com.flink.hbase.statemanagement.config.StateBackendConfig
            .configureDevRocksDB(env);
        
        // 运行示例
        runRichFunctionStateExample(env);
        
        // 执行作业
        env.execute("Rich Function状态示例");
    }
} 